package com.system.ad.mysql.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.system.ad.mysql.TemplateHolder;
import com.system.ad.mysql.dto.BinlogRowData;
import com.system.ad.mysql.dto.TableTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 监听binlog,根据日志操作索引
 */
@Slf4j
@Component
public class AggregationListener implements BinaryLogClient.EventListener {

    //数据库名字
    private String dbName;
    // 数据表名
    private String tableName;
    //表关联对应当前处理方法
    private Map<String, IListener> listenerMap = new HashMap<>();
    //模板信息
    private TemplateHolder templateHolder;

    // Spring 在创建类实例时会自动解析构造函数上的 @Autowired 注解，并根据注解指定的类型（如 TemplateHolder）来注入对应的依赖对象，无需手动创建和管理依赖对象。
    @Autowired
    public AggregationListener(TemplateHolder templateHolder) {
        this.templateHolder = templateHolder;
    }


    private String genKey(String dbName, String tableName) {
        return dbName + ":" + tableName;
    }

    /***
     * 注册方法，保存需要监听的数据表
     * @param _dbName
     * @param _tableName
     * @param iListener 接口
     */
    public void register(String _dbName, String _tableName, IListener iListener) {
        log.info("register :{}-{}", _dbName, _tableName);
        this.listenerMap.put(genKey(_dbName, _tableName), iListener);
    }


    @Override
    public void onEvent(Event event) {
        // 获取binlog日志信息
        EventType type = event.getHeader().getEventType();
        log.debug("event type:{}", type);
        if (type == EventType.TABLE_MAP) {
            TableMapEventData data = event.getData();
            this.tableName = data.getTable();
            this.dbName = data.getDatabase();
            return;

        }
        // 判断是否是更新、删除、写入操作
        if (type != EventType.EXT_UPDATE_ROWS && type != EventType.EXT_DELETE_ROWS && type != EventType.EXT_WRITE_ROWS) {
            return;
        }

        //判断表名与数据库名是否已赋值
        if (StringUtils.isEmpty(dbName) || StringUtils.isEmpty(tableName)) {
            return;
        }

        // 获取以保存的数据表
        String key = genKey(this.dbName, tableName);
        IListener iListener = this.listenerMap.get(key);
        //如果该表未注册，说明并不需要监听
        if (null == iListener) {
            log.debug("skip {}", key);
            return;
        }
        log.info("trigger event :{}", type.name());
        try {
            // 构建操作对象数据
            BinlogRowData binlogRowData = buildRowData(event.getData());
            if (binlogRowData == null) {
                return;
            }
            binlogRowData.setEventType(type);
            iListener.onEvent(binlogRowData);
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error(ex.getMessage());
        } finally {
            // 完成数据处理后清空数据
            this.dbName = "";
            this.tableName = "";
        }

    }

    /**
     * 根据binlog 数据操作类型获取对应值,统一返回数据格式
     * 操作类型不一样，操作值类型也不相同，需要转换
     * @param eventData
     * @return
     */
    private List<Serializable[]> getAfterValue(EventData eventData) {
        if (eventData instanceof WriteRowsEventData) {
            return ((WriteRowsEventData) eventData).getRows();
        }
        if (eventData instanceof UpdateRowsEventData) {
            // update的数据类型为map<k,v> value才是需要的数据
            ((UpdateRowsEventData) eventData).getRows().stream().map(Map.Entry::getValue)
                    .collect(Collectors.toList());
        }
        if (eventData instanceof DeleteRowsEventData) {
            return ((DeleteRowsEventData) eventData).getRows();
        }
        return Collections.emptyList();
    }

    private BinlogRowData buildRowData(EventData eventData) {
        // 根据表名获取表的操作对象
        TableTemplate table = templateHolder.getTable(tableName);
        if (null == table) {
            log.error("table{} not found", table);
            return null;
        }
        ArrayList<Map<String, String>> afterMapList = new ArrayList<>();
        // 获取操作值
        for (Serializable[] after : getAfterValue(eventData)) {
            HashMap<String, String> afterMap = new HashMap<>();
            int collen = after.length;
            for (int i = 0; i < collen; i++) {
                // 取出当前位置对应的列名
                String colName = table.getPosMap().get(i);
                if (null == colName) {
                    log.error("ignore position :{}", i);
                    continue;
                }
                String colValue = after[i].toString();
                afterMap.put(colName, colValue);
            }
            afterMapList.add(afterMap);
        }
        BinlogRowData binlogRowData = new BinlogRowData();
        binlogRowData.setAfter(afterMapList);
        binlogRowData.setTable(table);
        return binlogRowData;
    }
}
